package com.gugee.ins.data.collector.blogger.service.impl;

import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.constant.RedisKeyConst;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerRelationDao;
import com.gugee.ins.common.dal.spider.InsBloggerSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.InsBlogger;
import com.gugee.ins.common.model.mysql.blogger.InsBloggerRelation;
import com.gugee.ins.data.collector.blogger.service.InsBloggerCollectService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.collector.blogger.config.CollectorBloggerConfig;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * insBlogger服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsBloggerCollectServiceImpl implements InsBloggerCollectService {

    /**
     * 配置
     */
    @Autowired
    CollectorBloggerConfig collectorBloggerConfig;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     * redis
     */
    @Autowired
    RedisTemplate<String,String> redisTemplate;

    /**
     * 爬虫接口
     */
    @Autowired
    InsBloggerSpiderClient insBloggerSpiderClient;

    /**
     *
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    @Autowired
    InsBloggerRelationDao insBloggerRelationDao;

    /**
     * 处理insBlogger收入任务
     * @param insBloggerMessages
     */
    @Override
    public void handleInsBloggerCollect(List<InsBloggerMessage> insBloggerMessages) {
        if(!insBloggerMessages.isEmpty()){
            //去重
            List<Long> uids = insBloggerMessages.stream()
                    .mapToLong(InsBloggerMessage::getUid).distinct().boxed().collect(Collectors.toList());
            Set<Long> uidExists = new HashSet<>();
            HashMap<Long,InsBlogger> insBloggerHashMap = new HashMap<>(insBloggerMessages.size());
            int batchSearchSize = collectorBloggerConfig.getInsBloggerCollectBatchNums();
            int lastIndex = 0;
            List<InsBlogger> findBloggers;

            for(int i = batchSearchSize; i < uids.size() ; i += batchSearchSize){
                findBloggers = insBloggerDao.findExistUidByUids(uids.subList(lastIndex,i));
                lastIndex = i;
                if(findBloggers != null && !findBloggers.isEmpty()){
                    findBloggers.forEach(blogger -> {
                        uidExists.add(blogger.getUid());
                        insBloggerHashMap.put(blogger.getUid(),blogger);
                    });
                }
            }
            if(lastIndex < uids.size()){
                findBloggers = insBloggerDao.findExistUidByUids(uids.subList(lastIndex,uids.size()));
                if(findBloggers != null && !findBloggers.isEmpty()){
                    findBloggers.forEach(blogger -> {
                        uidExists.add(blogger.getUid());
                        insBloggerHashMap.put(blogger.getUid(),blogger);
                    });
                }
            }

            List<InsBloggerMessage> ibms = insBloggerMessages.stream().filter( msg -> {
               if(uidExists.contains(msg.getUid())){
  //为适应qps不够，队列容易积压，且不同的渠道来源区分不同的优先级使用uid接口。故此处最好注释
//                   if(insBloggerHashMap.get(msg.getUid()).getFollowerCount()==0 && !insBloggerHashMap.get(msg.getUid()).getIsZombie()){
//                       //发送更新消息
//                       sendUpdateBloggerMsg(msg);
//                   }

                   return false;
               }
               return true;
            }).collect(Collectors.toList());

            if(!ibms.isEmpty()){
                try {
                    insBloggerDao.batchInsert(ibms);
                    long nowTime = System.currentTimeMillis() / 1000;

                    List<InsBloggerRelation> insBloggerRelationList = new ArrayList<>(ibms.size());
                    for(InsBloggerMessage insBloggerMessage:ibms ){
                        if(insBloggerMessage.getTypeCode()>=300 && insBloggerMessage.getTypeCode()<=304){
                            InsBloggerRelation insBloggerRelation=new InsBloggerRelation();
                            insBloggerRelation.setUid(insBloggerMessage.getUid());
                            if(insBloggerMessage.getTypeCode()==InsBloggerMsgTypeCodeEnum.CLP_SRC_ACTIVE_FOLLOWER.getCode()){
                                insBloggerRelation.setIsActiveFollower(true);
                            }

                            if(insBloggerMessage.getTypeCode()==InsBloggerMsgTypeCodeEnum.CLP_SRC_MEDIA_TAG_PRODUCT.getCode()){
                                insBloggerRelation.setIsHasMerchantMedia(true);
                            }

                            if(insBloggerMessage.getTypeCode()==InsBloggerMsgTypeCodeEnum.CLP_SRC_MEDIA_SPONSOR.getCode()){
                                insBloggerRelation.setIsSponsor(true);
                            }

                            if(insBloggerMessage.getTypeCode()==InsBloggerMsgTypeCodeEnum.CLP_SRC_MEDIA_TAGGER.getCode()){
                                insBloggerRelation.setIsTaggedByMedia(true);
                            }

                            if(insBloggerMessage.getTypeCode()==InsBloggerMsgTypeCodeEnum.CLP_SRC_MERCHANT.getCode()){
                                insBloggerRelation.setIsMerchant(true);
                            }

                            insBloggerRelation.setCreateTime(nowTime);
                            insBloggerRelationList.add(insBloggerRelation);
                        }
                    }
                    if(!insBloggerRelationList.isEmpty()){
                        insBloggerRelationDao.batchInsert(insBloggerRelationList);
                    }

                } catch (Exception ex) {
                    ex.printStackTrace();
                    log.error("InsBloggerCollect batch insert error.  {}" ,  ex.getMessage());
                }

                ibms.forEach(insBloggerMessage -> {
                    try{
                        //删除redis key
                        deleteInsBloggerCollectRedisKey(insBloggerMessage.getUid());
                        //发送更新消息
                        insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_COLLECT.getCode());
                        sendUpdateBloggerMsg(insBloggerMessage);
                    }catch (Exception e){
                        log.error("delete ins blogger collect redis key error msg : {}",e.getMessage());
                    }
                });
            }

        }
    }

    /**
     * 发送博主收录的消息
     * @param uid
     * @param userName
     * @param picUrl
     */
    @Override
    public void sendCollectBloggerMsg(long uid,String userName,String picUrl,boolean isPrivate){
        if(uid > 0 && checkCollecting(uid)){
            //这里需要添加验证
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setUid(uid);
            insBloggerMessage.setUserName(userName);
            insBloggerMessage.setPicUrl(StringUtils.isEmpty(picUrl) ? "" : picUrl);
            insBloggerMessage.setRetryLimitTimes(collectorBloggerConfig.getInsBloggerCollectRetryTimes());
            insBloggerMessage.setRetryTimes(collectorBloggerConfig.getInsBloggerCollectRetryTimes());
            insBloggerMessage.setRanking(isPrivate);

            kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerCollectOnlineTopic(),insBloggerMessage);
        }
    }

    /**
     * 发送爬取关注人的消息
     */
    private void sendUpdateBloggerMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        insBloggerMessage.setRetryLimitTimes(collectorBloggerConfig.getInsBloggerUpdateRetryTimes());
        insBloggerMessage.setRetryTimes(collectorBloggerConfig.getInsBloggerUpdateRetryTimes());
        kafkaTemplateInsBlogger.send(collectorBloggerConfig.getInsBloggerUpdateOnlineTopic(),insBloggerMessage);
    }

    /**
     * 判断是否需要进行采集
     * @param uid
     * @return
     */
    @Override
    public boolean checkCollecting(long uid){
        //todo 这里后续可能需要进行修改，队列数据多了以后redis内存会剧增
        //查询redis是否正在被采集
        if(!StringUtils.isEmpty(redisTemplate.opsForValue().get(RedisKeyConst.INS_BLOGGER_COLLECTING + uid))){
            return false;
        }
        //查询数据库是否已存在
        val insBloggerOptional = insBloggerDao.findByUid(uid);
        if(insBloggerOptional != null){
            return false;
        }else{
            //再次查询redis是否正在被采集
            Boolean isExist = redisTemplate.opsForValue().setIfAbsent(RedisKeyConst.INS_BLOGGER_COLLECTING + uid,
                    (System.currentTimeMillis() / 1000) + "", collectorBloggerConfig.getAcceptInsBloggerCollectMsgTimeMinRange(), TimeUnit.MINUTES);
            if(isExist != null && !isExist){
                return false;
            }
        }
        return true;
    }

    /**
     * 删除key
     * @param uid
     */
    @Override
    public void deleteInsBloggerCollectRedisKey(long uid){
        if(uid <= 0){
            return;
        }
        redisTemplate.delete(RedisKeyConst.INS_BLOGGER_COLLECTING + uid);
    }
}
